package com.mxd.flink.connector.operator;

import com.mxd.flink.connector.config.LuaCmd;
import org.apache.commons.lang3.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import redis.clients.jedis.*;

import java.io.IOException;
import java.util.*;
import java.util.concurrent.*;

/**
 * @author rongdi
 * @date 2022/9/19 17:06
 */
public class RedisSingleOperator implements RedisOperator{

    private static final Logger LOG = LoggerFactory.getLogger(RedisSingleOperator.class);

    private final ExecutorService executorService =
            new ThreadPoolExecutor(Runtime.getRuntime().availableProcessors(), Runtime.getRuntime().availableProcessors(),
                    0L, TimeUnit.MILLISECONDS,
                    new ArrayBlockingQueue<>(10000,true));

    private transient JedisPool jedisPool;

    private transient JedisSentinelPool jedisSentinelPool;

    public RedisSingleOperator(final JedisPool jedisPool) {
        Objects.requireNonNull(jedisPool, "Jedis Pool can not be null");
        this.jedisPool = jedisPool;
        this.jedisSentinelPool = null;
    }

    public RedisSingleOperator(final JedisSentinelPool sentinelPool) {
        Objects.requireNonNull(sentinelPool, "Jedis Sentinel Pool can not be null");
        this.jedisPool = null;
        this.jedisSentinelPool = sentinelPool;
    }

    @Override
    public void close() throws IOException {
        if (this.jedisPool != null) {
            this.jedisPool.close();
        }
        if (this.jedisSentinelPool != null) {
            this.jedisSentinelPool.close();
        }
    }

    @Override
    public void hset(final String key, final String hashField, final String value) {
        Jedis jedis = null;
        try {
            jedis = this.getInstance();
            jedis.hset(key, hashField, value);
        }
        catch (Exception e) {
            if (LOG.isErrorEnabled()) {
                LOG.error("Cannot send Redis message with command HSET to key {} and hashField {} error message {}", key, hashField, e.getMessage());
            }
            throw e;
        }
        finally {
            this.releaseInstance(jedis);
        }
    }

    @Override
    public void rpush(final String listName, final String value) {
        Jedis jedis = null;
        try {
            jedis = this.getInstance();
            jedis.rpush(listName, value);
        }
        catch (Exception e) {
            if (LOG.isErrorEnabled()) {
                LOG.error("Cannot send Redis message with command RPUSH to list {} error message {}", listName, e.getMessage());
            }
            throw e;
        }
        finally {
            this.releaseInstance(jedis);
        }
    }

    @Override
    public void rpush(String listName, List<String> values) {
        Jedis jedis = null;
        try {
            jedis = this.getInstance();
            jedis.rpush(listName,values.toArray(new String[values.size()]));
        }
        catch (Exception e) {
            if (LOG.isErrorEnabled()) {
                LOG.error("Cannot send Redis message with command rpush to list {} error message {}", listName, e.getMessage());
            }
            throw e;
        }
        finally {
            this.releaseInstance(jedis);
        }
    }

    @Override
    public void lpush(final String listName, final String value) {
        Jedis jedis = null;
        try {
            jedis = this.getInstance();
            jedis.lpush(listName, value);
        }
        catch (Exception e) {
            if (LOG.isErrorEnabled()) {
                LOG.error("Cannot send Redis message with command LUSH to list {} error message {}", listName, e.getMessage());
            }
            throw e;
        }
        finally {
            this.releaseInstance(jedis);
        }
    }

    @Override
    public void lpush(String listName, List<String> values) {
        Jedis jedis = null;
        try {
            jedis = this.getInstance();
            jedis.lpush(listName,values.toArray(new String[values.size()]));
        }
        catch (Exception e) {
            if (LOG.isErrorEnabled()) {
                LOG.error("Cannot send Redis message with command rpush to list {} error message {}", listName, e.getMessage());
            }
            throw e;
        }
        finally {
            this.releaseInstance(jedis);
        }
    }

    @Override
    public List<String> brpop(final String listName) {
        List<String> value = null;
        Jedis jedis = null;
        try {
            jedis = this.getInstance();
            value = jedis.brpop(listName);
        }
        catch (Exception e) {
            if (LOG.isErrorEnabled()) {
                LOG.error("Cannot send Redis message with command brpop to list {} error message {}", listName, e.getMessage());
            }
            throw e;
        }
        finally {
            this.releaseInstance(jedis);
        }
        return value;
    }

    @Override
    public List<Object> rpop(String listName, long batchFetchRows) {
        List<Object> value = null;
        Jedis jedis = null;
        try {
            jedis = this.getInstance();
            String sha = LuaCmd.get(LuaCmd.BATCH_RPOP);
            if(StringUtils.isEmpty(sha)) {
                sha = jedis.scriptLoad(LuaCmd.BATCH_RPOP_LUA);
            }
            value = (List<Object>)jedis.evalsha(sha, Arrays.asList(listName),Arrays.asList(String.valueOf(batchFetchRows)));
        }
        catch (Exception e) {
            if (LOG.isErrorEnabled()) {
                LOG.error("Cannot send Redis message with command batch rpop to list {},size {} error message {}", listName,batchFetchRows, e.getMessage());
            }
            throw e;
        }
        finally {
            this.releaseInstance(jedis);
        }
        return value;
    }

    @Override
    public List<String> blpop(final String listName) {
        List<String> value = null;
        Jedis jedis = null;
        try {
            jedis = this.getInstance();
            value = jedis.blpop(1000, listName);
        }
        catch (Exception e) {
            if (LOG.isErrorEnabled()) {
                LOG.error("Cannot send Redis message with command blpop to list {} error message {}", listName, e.getMessage());
            }
            throw e;
        }
        finally {
            this.releaseInstance(jedis);
        }
        return value;
    }

    @Override
    public List<Object> lpop(String listName, long batchFetchRows) {
        List<Object> value = null;
        Jedis jedis = null;
        try {
            jedis = this.getInstance();
            String sha = LuaCmd.get(LuaCmd.BATCH_LPOP);
            if(StringUtils.isEmpty(sha)) {
                sha = jedis.scriptLoad(LuaCmd.BATCH_LPOP_LUA);
            }
            value = (List<Object>)jedis.evalsha(sha, Arrays.asList(listName),Arrays.asList(String.valueOf(batchFetchRows)));
        }
        catch (Exception e) {
            if (LOG.isErrorEnabled()) {
                LOG.error("Cannot send Redis message with command batch lpop to list {},size {} error message {}", listName,batchFetchRows, e.getMessage());
            }
            throw e;
        }
        finally {
            this.releaseInstance(jedis);
        }
        return value;
    }

    @Override
    public void sadd(final String setName, final String value) {
        Jedis jedis = null;
        try {
            jedis = this.getInstance();
            jedis.sadd(setName, value);
        }
        catch (Exception e) {
            if (LOG.isErrorEnabled()) {
                LOG.error("Cannot send Redis message with command sadd {} error message {}", setName, e.getMessage());
            }
            throw e;
        }
        finally {
            this.releaseInstance(jedis);
        }
    }

    @Override
    public void sadd(final String setName, final List<String> values) {
        Jedis jedis = null;
        try {
            jedis = this.getInstance();
            jedis.sadd(setName, values.toArray(new String[values.size()]));
        }
        catch (Exception e) {
            if (LOG.isErrorEnabled()) {
                LOG.error("Cannot send Redis message with command sadd {} size {} error message {}", setName, values.size(), e.getMessage());
            }
            throw e;
        }
        finally {
            this.releaseInstance(jedis);
        }
    }

    @Override
    public Set<String> spop(final String setName, final long batchFetchRows) {
        Jedis jedis = null;
        try {
            jedis = this.getInstance();
            return jedis.spop(setName,batchFetchRows);
        }
        catch (Exception e) {
            if (LOG.isErrorEnabled()) {
                LOG.error("Cannot send Redis message with command RPUSH to set {} error message {}", setName, e.getMessage());
            }
            throw e;
        }
        finally {
            this.releaseInstance(jedis);
        }
    }

    @Override
    public void set(final String key, final String value) {
        Jedis jedis = null;
        try {
            jedis = this.getInstance();
            jedis.set(key, value);
        }
        catch (Exception e) {
            if (LOG.isErrorEnabled()) {
                LOG.error("Cannot send Redis message with command SET to key {} error message {}", key, e.getMessage());
            }
            throw e;
        }
        finally {
            this.releaseInstance(jedis);
        }
    }

    @Override
    public void pfadd(final String key, final String element) {
        Jedis jedis = null;
        try {
            jedis = this.getInstance();
            jedis.pfadd(key, element);
        }
        catch (Exception e) {
            if (LOG.isErrorEnabled()) {
                LOG.error("Cannot send Redis message with command PFADD to key {} error message {}", key, e.getMessage());
            }
            throw e;
        }
        finally {
            this.releaseInstance(jedis);
        }
    }

    @Override
    public void zadd(final String key, final String score, final String element) {
        Jedis jedis = null;
        try {
            jedis = this.getInstance();
            jedis.zadd(key, Double.valueOf(score), element);
        }
        catch (Exception e) {
            if (LOG.isErrorEnabled()) {
                LOG.error("Cannot send Redis message with command ZADD to set {} error message {}", key, e.getMessage());
            }
            throw e;
        }
        finally {
            this.releaseInstance(jedis);
        }
    }

    @Override
    public void zrem(final String key, final String element) {
        Jedis jedis = null;
        try {
            jedis = this.getInstance();
            jedis.zrem(key, element);
        }
        catch (Exception e) {
            if (LOG.isErrorEnabled()) {
                LOG.error("Cannot send Redis message with command ZREM to set {} error message {}", key, e.getMessage());
            }
            throw e;
        }
        finally {
            this.releaseInstance(jedis);
        }
    }

    private Jedis getInstance() {
        if (this.jedisSentinelPool != null) {
            return this.jedisSentinelPool.getResource();
        }
        return this.jedisPool.getResource();
    }

    private void releaseInstance(final Jedis jedis) {
        if (jedis == null) {
            return;
        }
        try {
            jedis.close();
        }
        catch (Exception e) {
            LOG.error("Failed to close (return) instance to pool", e);
        }
    }

    @Override
    public void setbit(final String key, final long value, final boolean offset) {
        Jedis jedis = null;
        try {
            jedis = this.getInstance();
            jedis.setbit(key, value, offset);
        }
        catch (Exception e) {
            if (LOG.isErrorEnabled()) {
                LOG.error("Cannot send Redis message with command setbit to key {} with value {} and offset {} error message {}", key, value, offset, e.getMessage());
            }
            throw e;
        }
        finally {
            this.releaseInstance(jedis);
        }
    }

    @Override
    public boolean getbit(final String key, final long offset) {
        Jedis jedis = null;
        boolean result = false;
        try {
            jedis = this.getInstance();
            result = jedis.getbit(key, offset);
        }
        catch (Exception e) {
            if (LOG.isErrorEnabled()) {
                LOG.error("Cannot send Redis message with command getbit to key {} with offset {} error message {}", key, offset, e.getMessage());
            }
            throw e;
        }
        finally {
            this.releaseInstance(jedis);
        }
        return result;
    }

    @Override
    public String hget(final String key, final String field) {
        Jedis jedis = null;
        String result = null;
        try {
            jedis = this.getInstance();
            result = jedis.hget(key, field);
        }
        catch (Exception e) {
            if (LOG.isErrorEnabled()) {
                LOG.error("Cannot send Redis message with command hget to key {} with field {} error message {}", key, field, e.getMessage());
            }
            throw e;
        }
        finally {
            this.releaseInstance(jedis);
        }
        return result;
    }

    @Override
    public CompletableFuture<String> asyncHget(String key, String field) {
        CompletableFuture<String> completableFuture = new CompletableFuture<>();
        executorService.submit(() -> {
            completableFuture.complete(hget(key, field));
        });
        return completableFuture;
    }

    @Override
    public Map<String, String> hgetAll(final String key) {
        Jedis jedis = null;
        Map<String, String> result = null;
        try {
            jedis = this.getInstance();
            result = jedis.hgetAll(key);
        }
        catch (Exception e) {
            if (LOG.isErrorEnabled()) {
                LOG.error("Cannot send Redis message with command hget to key {} error message {}", key, e.getMessage());
            }
            throw e;
        }
        finally {
            this.releaseInstance(jedis);
        }
        return result;
    }

    @Override
    public CompletableFuture<Map<String, String>> asyncHgetAll(String key) {
        CompletableFuture<Map<String, String>> completableFuture = new CompletableFuture<>();
        executorService.submit(() -> {
            completableFuture.complete(hgetAll(key));
        });
        return completableFuture;
    }

    @Override
    public String get(final String key) {
        Jedis jedis = null;
        String result = null;
        try {
            jedis = this.getInstance();
            result = jedis.get(key);
        }
        catch (Exception e) {
            if (LOG.isErrorEnabled()) {
                LOG.error("Cannot send Redis message with command get to key {} error message {}", key, e.getMessage());
            }
            throw e;
        }
        finally {
            this.releaseInstance(jedis);
        }
        return result;
    }

    @Override
    public CompletableFuture<String> asyncGet(String key) {
        CompletableFuture<String> completableFuture = new CompletableFuture<>();
        executorService.submit(() -> {
            completableFuture.complete(get(key));
        });
        return completableFuture;
    }

    @Override
    public void hdel(final String key, final String field) {
        Jedis jedis = null;
        try {
            jedis = this.getInstance();
            jedis.hdel(key, field);
        }
        catch (Exception e) {
            if (LOG.isErrorEnabled()) {
                LOG.error("Cannot send Redis message with command hget to key {} with field {} error message {}", key, field, e.getMessage());
            }
            throw e;
        }
        finally {
            this.releaseInstance(jedis);
        }
    }

    @Override
    public void del(final String key) {
        Jedis jedis = null;
        try {
            jedis = this.getInstance();
            jedis.del(key);
        }
        catch (Exception e) {
            if (LOG.isErrorEnabled()) {
                LOG.error("Cannot send Redis message with command del to key {} error message {}", key, e.getMessage());
            }
            throw e;
        }
        finally {
            this.releaseInstance(jedis);
        }
    }

    @Override
    public Long expire(final String key, final int seconds) {
        Jedis jedis = null;
        try {
            jedis = this.getInstance();
            jedis.expire(key, seconds);
        }
        catch (Exception e) {
            if (LOG.isErrorEnabled()) {
                LOG.error("Cannot send Redis message with command exists to key {}  seconds {} error message {}", key, seconds, e.getMessage());
            }
            throw e;
        }
        finally {
            this.releaseInstance(jedis);
        }
        return null;
    }

    @Override
    public boolean sismember(final String key, final String member) {
        Jedis jedis = null;
        boolean result = false;
        try {
            jedis = this.getInstance();
            result = jedis.sismember(key, member);
        }
        catch (Exception e) {
            if (LOG.isErrorEnabled()) {
                LOG.error("Cannot send Redis message with command exists to key {}  member {} error message {}", key, member, e.getMessage());
            }
            throw e;
        }
        finally {
            this.releaseInstance(jedis);
        }
        return result;
    }

    @Override
    public long scard(final String key) {
        Jedis jedis = null;
        long result;
        try {
            jedis = this.getInstance();
            result = jedis.scard(key);
        }
        catch (Exception e) {
            if (LOG.isErrorEnabled()) {
                LOG.error("Cannot send Redis message with command scard to key {} error message {}", key, e.getMessage());
            }
            throw e;
        }
        finally {
            this.releaseInstance(jedis);
        }
        return result;
    }

    @Override
    public void srem(final String setName, final String value) {
        Jedis jedis = null;
        try {
            jedis = this.getInstance();
            jedis.srem(setName, value);
        }
        catch (Exception e) {
            if (LOG.isErrorEnabled()) {
                LOG.error("Cannot send Redis message with command srem to setName {} value : {} error message {}", setName, value, e.getMessage());
            }
            throw e;
        }
        finally {
            this.releaseInstance(jedis);
        }
    }

    @Override
    public long incrByLong(final String key, final long value) {
        Jedis jedis = null;
        long result = 0L;
        try {
            jedis = this.getInstance();
            result = jedis.incrBy(key, value);
        }
        catch (Exception e) {
            if (LOG.isErrorEnabled()) {
                LOG.error("Cannot send Redis message with command incrBy to key {} and value {} error message {}", key, value, e.getMessage());
            }
            throw e;
        }
        finally {
            this.releaseInstance(jedis);
        }
        return result;
    }

    @Override
    public void hset(final String key, final String hashField, final String value, final Integer expireTime) {
        Jedis jedis = null;
        try {
            jedis = this.getInstance();
            jedis.hset(key, hashField, value);
            if (expireTime != null) {
                jedis.expire(key, expireTime);
            }
        }
        catch (Exception e) {
            if (LOG.isErrorEnabled()) {
                LOG.error("Cannot send Redis message with command HSET to key {} and hashField {} error message {}", key, hashField, e.getMessage());
            }
            throw e;
        }
        finally {
            this.releaseInstance(jedis);
        }
    }
}
